/*
 *
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 */

package com.qingyan.flink.datasource.mysql;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.qingyan.flink.core.properties.FlinkDataSourceProperties;
import com.qingyan.flink.core.source.cdc.AbstractCDCBuilder;
import com.qingyan.flink.core.source.cdc.CDCBuilder;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * MysqlCDCBuilder
 *
 * @author wenmo
 * @since 2022/4/12 21:29
 **/
public class MysqlCDCBuilder extends AbstractCDCBuilder {

    public static final String KEY_WORD = "mysql-cdc";
    private static final String METADATA_TYPE = "MySql";

    private MySqlDatasourceProperties datasourceProperties;


    public MysqlCDCBuilder(FlinkDataSourceProperties sourceProperties, Map<String, String> sourceConfig) {
        super(sourceProperties, sourceConfig);
        datasourceProperties = new MySqlDatasourceProperties();
        BeanUtil.fillBeanWithMap(sourceProperties.getProperties(), datasourceProperties, true);
    }

    @Override
    public String getHandle() {
        return KEY_WORD;
    }

    @Override
    public CDCBuilder create(FlinkDataSourceProperties properties, Map<String, String> sourceConfig) {
        return new MysqlCDCBuilder(properties, sourceConfig);
    }

    @Override
    public DataStreamSource<String> build(StreamExecutionEnvironment env) {
        String[] databases = super.getSchemaList().toArray(new String[0]);
        String serverId = sourceConfig.get("server-id");
        String serverTimeZone = sourceConfig.get("server-time-zone");
        String fetchSize = sourceConfig.get("scan.snapshot.fetch.size");
        String connectTimeout = sourceConfig.get("connect.timeout");
        String connectMaxRetries = sourceConfig.get("connect.max-retries");
        String connectionPoolSize = sourceConfig.get("connection.pool.size");
        String heartbeatInterval = sourceConfig.get("heartbeat.interval");
        String chunkSize = sourceConfig.get("scan.incremental.snapshot.chunk.size");
        Properties debeziumProperties = new Properties();
        // 为部分转换添加默认值
        debeziumProperties.setProperty("bigint.unsigned.handling.mode", "long");
        debeziumProperties.setProperty("decimal.handling.mode", "string");

        MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder()
                .hostname(datasourceProperties.getHostname())
                .port(datasourceProperties.getPort())
                .username(datasourceProperties.getUsername())
                .password(datasourceProperties.getPassword());
        sourceBuilder.databaseList(databases);
        List<String> schemaTableNameList = super.getSchemaTableList();
        if (CollUtil.isNotEmpty(schemaTableNameList)) {
            sourceBuilder.tableList(schemaTableNameList.toArray(new String[0]));
        } else {
            sourceBuilder.tableList();
        }
        sourceBuilder.deserializer(new MysqlJsonDebeziumDeserializationSchema());
        sourceBuilder.debeziumProperties(debeziumProperties);
        if (StrUtil.isNotEmpty(serverId)) {
            sourceBuilder.serverId(serverId);
        }
        if (StrUtil.isNotEmpty(serverTimeZone)) {
            sourceBuilder.serverTimeZone(serverTimeZone);
        }
        if (StrUtil.isNotEmpty(fetchSize)) {
            sourceBuilder.fetchSize(Integer.valueOf(fetchSize));
        }
        if (StrUtil.isNotEmpty(connectTimeout)) {
            sourceBuilder.connectTimeout(Duration.ofMillis(Long.valueOf(connectTimeout)));
        }
        if (StrUtil.isNotEmpty(connectMaxRetries)) {
            sourceBuilder.connectMaxRetries(Integer.valueOf(connectMaxRetries));
        }
        if (StrUtil.isNotEmpty(connectionPoolSize)) {
            sourceBuilder.connectionPoolSize(Integer.valueOf(connectionPoolSize));
        }
        if (StrUtil.isNotEmpty(heartbeatInterval)) {
            sourceBuilder.heartbeatInterval(Duration.ofMillis(Long.valueOf(heartbeatInterval)));
        }

        if (StrUtil.isNotEmpty(chunkSize)) {
            sourceBuilder.splitSize(Integer.parseInt(chunkSize));
        }

        return env.fromSource(sourceBuilder.build(), WatermarkStrategy.noWatermarks(), "MySQL CDC Source");
    }


    @Override
    public String getSchema() {
        return null;
    }

}
